package xin.marcher.module.message.mq.consumer.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import xin.marcher.module.common.message.DataChangeMessage;
import xin.marcher.module.common.redis.RedisLock;
import xin.marcher.module.message.domain.bo.DataChangeMessageConfigBO;
import xin.marcher.module.message.domain.bo.DataMessageBO;
import xin.marcher.module.message.domain.bo.DataSendMessageBO;
import xin.marcher.module.message.mq.producer.DataMessageProducer;
import xin.marcher.module.message.service.MessageService;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

/**
 * 外部消息监听
 * 背景说明及流程: 需要被外部系统消费的数据, 会先存入数据库, 看 {@link DataChangeListener} 消费流程,
 * 然后需要获取数据的系统发送消息到 MQ, 消息服务收到后会查询数据库的消息, 封装后再通过 MQ 发出去, 此时就可以通过 MQ 消费需要的数据了
 */
@Slf4j
@Component
public class DataExternalChangeListener implements MessageListenerConcurrently {

    @Resource
    private MessageService messageService;

    @Resource
    private DataMessageProducer dataMessageProducer;

    @Resource
    private RedisLock redisLock;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        for (MessageExt messageExt : list) {
            String messageNo = new String(messageExt.getBody());
            if (!redisLock.lock(messageNo)) {
                continue;
            }
            try {
                DataMessageBO dataMessageDetail = messageService.getDataMessageDetail(messageNo);
                // 未命中到外部消息的数据，默认不处理
                if (Objects.isNull(dataMessageDetail)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                // 获取外部消息的报文对象
                List<DataSendMessageBO> sendDataMessageList = messageService.getSendDataMessage(dataMessageDetail);
                if (CollectionUtils.isEmpty(sendDataMessageList)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 发送外部消息
                sendDataMessage(sendDataMessageList);

                // 删除这条记录
                messageService.deleteMessage(dataMessageDetail);
            } catch (Exception e) {
                log.error("consume error, 消费外部数据变更消息失败", e);
                // 本次消费失败，下次重新消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            } finally {
                redisLock.unlock(messageNo);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 发送外部的消息
     *
     * @param sendDataMessageList
     */
    private void sendDataMessage(List<DataSendMessageBO> sendDataMessageList) {
        for (DataSendMessageBO dataChangeMessage : sendDataMessageList) {
            DataChangeMessage dataMessage = dataChangeMessage.getDataChangeMessage();
            DataChangeMessageConfigBO dataChangeMessageConfigBO = dataChangeMessage.getDataChangeMessageConfigBO();
            // 发送一个延迟队列的消息出去
            dataMessageProducer.send(dataMessage, dataChangeMessageConfigBO.getMessageTopic(), dataChangeMessageConfigBO.getDelayLevel());
        }
    }
}
